package com.example.rpcdemo.core.client;

import com.example.rpcdemo.core.protocol.RpcRequest;
import com.example.rpcdemo.core.protocol.RpcResponse;
import com.example.rpcdemo.core.codec.RpcDecoder;
import com.example.rpcdemo.core.codec.RpcEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

public class RpcClient {
    private final String host;
    private final int port;
    private Channel channel;
    private final EventLoopGroup group = new NioEventLoopGroup();
    private final ConcurrentHashMap<String, RpcFuture> pendingRequests = new ConcurrentHashMap<>();
    
    public RpcClient(String host, int port) {
        this.host = host;
        this.port = port;
    }
    
    public void connect() throws Exception {
        if (host == null || port == 0) {
            throw new IllegalStateException("Host and port must be provided.");
        }
        
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline()
                                .addLast(new RpcDecoder(RpcResponse.class))
                                .addLast(new RpcEncoder())
                                .addLast(new RpcClientHandler(pendingRequests));
                    }
                });
        
        ChannelFuture future = bootstrap.connect(host, port).sync();
        this.channel = future.channel();
        System.out.println("RPC Client connected to " + host + ":" + port);
    }
    
    public Object sendRequest(RpcRequest request) throws Exception {
        RpcFuture future = new RpcFuture();
        pendingRequests.put(request.getClassName() + "." + request.getMethodName(), future);
        
        channel.writeAndFlush(request);
        
        return future.get(10, TimeUnit.SECONDS);
    }
    
    public void close() {
        if (channel != null) {
            channel.close();
        }
        group.shutdownGracefully();
    }
}